<?php
/**
 * RabbitMq 使用类库
 * 依赖：
 * Amqp 扩展
 * php-amqp 类库
 */

namespace app\common\lib\Mq;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
use app\common\lib\Mq\Drive;

class RabbitMq extends Drive
{
    static private $instance;
    static private $connection;
    static private $channel;
    const EXCHANGE_NAME = 'normal_direct';
    const DELAY_EXCHANGE = 'delay_exchange';


    private function __construct()
    {
        try {
            $config = config('rabbitmq');//载入配置文件
            self::$connection = new AMQPStreamConnection(
                $config['hostname'],
                $config['hostport'],
                $config['username'],
                $config['password'],
                $config['vhost']
            );
            self::$channel = self::$connection->channel();
            self::$channel->exchange_declare(self::EXCHANGE_NAME, 'direct', false, true, false);
        } catch (\Exception $exception) {
        }
    }

    /**
     * 检查是否已经连接
     * @return bool
     */
    public function isConnect()
    {
        if (!self::$connection) {
            return false;
        }
        if (!self::$connection->isConnected()) {
            return false;
        }
        if (!self::$channel) {
            return false;
        }
        return true;
    }

    /**
     * 实例化
     * @return MsgQueue
     */
    static function instance()
    {
        if (!self::$instance instanceof self) {
            self::$instance = new self();
        }
        return self::$instance;
    }

    /**
     * 发送消息
     * @param $queueName
     * @param $data
     */
    public function sendMessage($queueName, $data)
    {
        if (!$this->isConnect()) {
            return self::response(-1, '发送失败');
        }
        try {
            self::$channel->queue_declare($queueName, false, true, false, false);
            $msg = new AMQPMessage(
                $data,
                array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)
            );
            self::$channel->basic_qos(null, 1, null);
            self::$channel->basic_publish($msg, self::EXCHANGE_NAME, $queueName);
        } catch (\Exception $exception) {
            return self::response(-1, '发送失败');
        }
        return self::response(1, '发送成功');
    }

    /**
     * 发送延时消息
     * @param $queueName
     * @param $data
     * @param $delayScond
     * @return false|string
     * @date 2021-06-16 18:29
     */
    public function sendDelayMessage($queueName, $data, $delaySecond)
    {
        if (!$this->isConnect()) {
            return self::response(-1, '发送失败');
        }
        try {
            self::$channel->exchange_declare(self::DELAY_EXCHANGE, 'x-delayed-message', false, true, false, false, false, new AMQPTable(['x-delayed-type' => 'direct']));
            self::$channel->queue_declare($queueName, false, true, false, false);
            $msg = new AMQPMessage(
                $data,
                array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)
            );
            $msg->set('application_headers', new AMQPTable([
                'x-delay' => $delaySecond * 1000 // 延迟时间，单位毫秒
            ]));
            self::$channel->basic_qos(null, 1, null);
            self::$channel->basic_publish($msg, self::DELAY_EXCHANGE, $queueName);
        } catch (\Exception $exception) {
            return self::response(-1, '发送失败');
        }
        return self::response(1, '发送成功');
    }

    /**
     * 接受消息
     * @param $queueName
     */
    public function receiveMessage($queueName = [], $callback, $bindQueueName)
    {
        self::$channel->queue_declare($bindQueueName, false, true, false, false);
        foreach ($queueName as $qv) {
            self::$channel->queue_bind($bindQueueName, self::EXCHANGE_NAME, $qv);
        }
        /*$callback = function($msg){
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        };*/
        self::$channel->basic_qos(null, 1, null);
        self::$channel->basic_consume($bindQueueName, '', false, false, false, false, $callback);

        while (count(self::$channel->callbacks)) {
            self::$channel->wait();
        }
    }

    /**
     * Note:接收延时消息
     * Date:2021-09-06 14:49
     * @param array $queueName
     * @param $callback
     * @throws \ErrorException
     */
    public function receiveMessageDelay($queueName = [], $callback, $bindQueueName)
    {
        self::$channel->exchange_declare(self::DELAY_EXCHANGE, 'x-delayed-message', false, true, false, false, false, new AMQPTable(['x-delayed-type' => 'direct']));
        self::$channel->queue_declare($bindQueueName, false, true, false, false);
        // self::$channel->queue_declare($queue_name, false, true, false, false);
        foreach ($queueName as $qv) {
            self::$channel->queue_bind($bindQueueName, self::DELAY_EXCHANGE, $qv);
        }
        /*$callback = function($msg){
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        };*/
        self::$channel->basic_qos(null, 1, null);
        self::$channel->basic_consume($bindQueueName, '', false, false, false, false, $callback);

        while (count(self::$channel->callbacks)) {
            self::$channel->wait();
        }
    }

    /**
     * 销毁
     */
    public function __destruct()
    {
        if (self::$channel) {
            self::$channel->close();
            self::$connection->close();
        }
    }
}
